package com.pai4j.common.service.messagequeue.local;

import com.pai4j.common.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;

import java.util.List;
import java.util.Queue;

@Slf4j
public abstract class AbstractLocalQueue<Q extends Queue, M> {


    /**
     * 获取本地队列对象
     *
     * 由具体实现类来实现创建
     *
     * @return
     */
    protected abstract Queue<M> queue();


    /**
     * 实际对消任务消费，由实现类完成
     *
     * @param message
     * @return
     */
    protected abstract boolean consume(M message);

    /**
     * 队列写入异常拒绝策略实现
     *
     * @param messages
     * @return
     */
    public abstract boolean doReject(List<M> messages);


    /**
     * 异常/处理失败后是否需要重新归队
     *
     * @return
     */
    protected boolean backQueue() {
        return true;
    }

    /**
     * 入队
     *
     * @param message
     * @return
     */
    public boolean push(M message) {
        if (message == null) {
            return true;
        }
        return push(List.of(message));
    }

    public boolean push(List<M> messages) {
        if (CollectionUtils.isEmpty(messages)) {
            return true;
        }
        try {
            return queue().addAll(messages);
        } catch (Exception e) {
            log.error("本地队列入队异常, ", e);
            doReject(messages);
        }
        return true;
    }


    /**
     * 出队
     *
     * @return
     */
    public M execConsume() {
        while (true) {
            M message = null;
            try {
                message = (M) queue().poll();
                if (message != null) {
                    boolean res = consume(message);
                    if (!res && backQueue()) {
                        this.push(message);
                    }
                }
            } catch (Exception e) {
                log.error("通知消息队列消费异常！message:{}", JsonUtil.toJsonString(message), e);
                if (backQueue()) {
                    this.push(message);
                }
            }
        }
    }

}
